package com.nl.deal;

import java.io.IOException;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.druid.support.logging.Log;
import com.alibaba.druid.support.logging.LogFactory;
import com.nl.DealDataToSendMain;
import com.nl.redis.RedisCluster;
import com.nl.util.AES;
import com.nl.util.ConnectionManager;
import com.nl.util.StringUtil;


public class DealDataToSend implements Runnable {

	private static final Logger log = LoggerFactory.getLogger(DealDataToSend.class);

	

	private SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
	private SimpleDateFormat delaysdf = new SimpleDateFormat("yyyyMMddHHmmss");
	private String redisCfg;
	private int count=0;
	private int count1=0;
	private int count2 =0;
	private int count3=0;
	private RedisCluster cluster;

	private List<SmsSender> senders = new ArrayList<>();
	private String sendSql;
	private String senderNum;


	private SimpleDateFormat sdf1 = new SimpleDateFormat("yyyy-MM-dd HHmmss");
	private String flag;
	private String tableName;
	public DealDataToSend (String redisCfg,String senderNum,String flag,String tableName) {
		this.redisCfg = redisCfg;
		this.senderNum = senderNum;
		this.flag = flag;
		this.tableName = tableName;
	}

	@Override
	public void run() {
		log.info("deal data start!");
		try {
			cluster = new RedisCluster(redisCfg);
			sendSql = "insert into "+tableName+"(id,receiver,sender,content,result," +
					"create_date,modify_date,fetch_id,city_id,region_id" +
					",region_name,station_id,station_name,mgr_id,mgr_name,user_place,user_city,is_local,is_mark,city_ext_id,redis_count)" +
					"values(seq_bomc_cmpp.nextval,?,?,?,?,sysdate,sysdate,?,?,?,?,?,?,?,?,?,?,?,?,?,?)";


			while(true) {
				String msg = DataCache.dealData.poll();
				//			String msg = "OLnwNv9hdOc1s_I6G3-EFg,Kpd5a00p0uWCN56I0I9FXg,12,20201030172017,20201030172033,2000250,12,0,0";
				if (null != msg) {
					count1 ++;
					if (count1 %5000 ==0) {
						log.info("已处理数据"+count1+"");
					}
					try{
						List<String> list = StringUtil.split(msg, ",");
						deal( list,msg);
					} catch (Exception e) {
						log.error("deal data error,msg = {"+msg+"}",e);
					}
				}	
			}
		} catch (IOException e1) {
			e1.printStackTrace();
			log.info("deal data start error, msg = "+e1.getMessage());
		}

	}

	private void deal(List<String> list,String msg) throws ParseException {

		String msisdn = list.get(0);
		String cityId = list.get(2);
		String blackUser = DataCache.blackUser.get(msisdn);
		///黑名单和七日已发可以提前过滤，不依赖其他条件
		///过滤黑名单
		//if (null == blackUser) {
		String stationId = list.get(1);
		String mgrIds = DataCache.stationMgrRelationMap.get(stationId);
		if (null != mgrIds && !"".equals(mgrIds)) {
			List<String> mgrIdCollection = StringUtil.split(mgrIds, ",");
			for (String mgrId: mgrIdCollection) {
				///过滤七日已发
				//					log.debug("get key = "+msisdn+"#"+mgrId+",map = "+map);
				//					log.debug("mgrIdCollection = "+mgrIdCollection+",deal msg = "+msg);
				if (!DataCache.todaySendMsisdns.containsKey(msisdn+"#"+mgrId)) {
					if (!cluster.existKey(msisdn+"#"+mgrId)) {
						//						log.debug("is not sevenuser,mgrId = "+mgrId+",deal msg = "+msg);
						SmsTemplateAttr smsTemplateAttr = DataCache.smsTemplateAttrMap.get(mgrId);
						if (null == smsTemplateAttr) {
							log.info("mgrId = "+mgrId);
							continue;
						}

						//获取项目标识，进行黑名单过滤
						String mgrFlag = smsTemplateAttr.getFlag();
						if("2".equals(mgrFlag)&&null!=blackUser){
							log.info("mgrId = "+mgrId+",phone = "+msisdn+",this is xuYiMgr blackUser,pass!");
							continue;
						}
						String now = sdf.format(new Date());
						String isLocal = list.get(7);
						String isSendLocal = smsTemplateAttr.getIs_local();
						///从原视图choose_file_data剥离的逻辑，短信审核状态、启停状态、发送时间、是否发送本地号码过滤
						///短信审核状态、启停状态、发送时间已在sql中做了筛选
						if (now.compareTo(smsTemplateAttr.getStart_date()) >= 0 
								&& now.compareTo(smsTemplateAttr.getEnd_date()) <= 0 
								) {
							///&& isLocal.compareTo(isSendLocal) >= 0
							if (isSendLocal.equals("2")) {
								isSendLocal = isLocal;
							}
							//IT传来的isLocal 0是本地，1是外地 和本地不一致则直接过滤掉
							if (!isLocal.equals(isSendLocal)) {
								log.debug("isLocal = "+isLocal+",isSendLocal = "+isSendLocal+", msgId = "+mgrId);
								continue;
							}

							String isCz = list.get(8);
							///剔除常驻人口
							if ("0".equals(isCz) || "0".equals(isSendLocal)) {
								ThresholdInfo info = DataCache.thresholdMap.get(mgrId);
								if (info == null) {
									continue;
								}
								/*///判断不同周期的阀值，阀值过滤
								String count = cluster.get(mgrId);
								if (null == count || "".equals(count)) {
									count = "0";
								}
								int dayCount = Integer.valueOf(count);*/

								StationMgrRelation stationMgrRelation = DataCache.stationMgrINfoRelationMap.get(mgrId);
								String regionId = null;
								String regionName = null;
								String stationName = null;
								String mgrName = null;
								String cityExtId = null;
								if (null != stationMgrRelation) {
									regionId = stationMgrRelation.getRegion_id();
									regionName = stationMgrRelation.getRegion_name();
									stationName = stationMgrRelation.getStation_name();
									mgrName = stationMgrRelation.getName();
									cityExtId = stationMgrRelation.getCity_id();
								}

								String key = msisdn+"#"+mgrId;


								try{
									//String result = cluster.hmsetResult(key, sevenMap);
									Long result=cluster.hsetNx(key, "INSERT_FLAG", flag);
									if("0".equals(String.valueOf(result))){
										//log.info("----key has already exist!--msisdn+mgrId:"+key);
										continue;									
									}

									cluster.hsetNx(key, "ARRIVAL_CITY", cityId);
									cluster.hsetNx(key, "SEND_TIME", now);
									cluster.hsetNx(key, "REGION_ID", regionId);
									cluster.hsetNx(key, "IS_LOCAL", isLocal);

									/*long timeCount=0;

								long weekCount=0;
								long monthCount=0;
								long quarterCount=0;
								long yearCount=0;*/

									long timeCount=0;
									long dayCount=0;
									long weekCount=0;
									long monthCount=0;
									long quarterCount=0;
									long yearCount=0;
									long totalCount=0;

									//redis中计算先做加一操作
									//timeCount=cluster.incrBy(mgrId+"timeCount", 1L);
									dayCount=cluster.incrBy(mgrId+"dayCount", 1L);
									weekCount=cluster.incrBy(mgrId+"weekCount", 1L);
									monthCount=cluster.incrBy(mgrId+"monthCount", 1L);
									quarterCount=cluster.incrBy(mgrId+"quarterCount", 1L);
									yearCount=cluster.incrBy(mgrId+"yearCount", 1L);
									totalCount=cluster.incrBy(mgrId+"totalCount", 1L);
									//log.info("----dayCount----"+dayCount);
									//log.info("-----info----"+info.toString());
									//阀值过滤
									if ((info.getTotalThreshold() != 0 && info.getTotalThreshold() < totalCount)
											|| (info.getCycleThreshold() != 0 && info.getCycle() == 4 && info.getCycleThreshold() < dayCount)
											) {								
										//条件不符合，数量做减一操作。
										//cluster.decrBy(mgrId+"timeCount", 1L);
										cluster.decrBy(mgrId+"dayCount", 1L);
										cluster.decrBy(mgrId+"weekCount", 1L);
										cluster.decrBy(mgrId+"monthCount", 1L);
										cluster.decrBy(mgrId+"quarterCount", 1L);
										cluster.decrBy(mgrId+"yearCount", 1L);
										cluster.decrBy(mgrId+"totalCount", 1L);
										//log.info("项目发送量已经到达阀值,已做减一操作"+mgrId+"目前发送量为："+(dayCount-1));
										//发送量已到，将redis中key过期时间设置为0,不记入七日已发
										cluster.expire(key, 0);
										continue;
									}								

									SmsSender smsSender = new SmsSender();
									smsSender.setReceiver(msisdn);
									smsSender.setSender(senderNum);
									smsSender.setContent(DataCache.msgContens.get(mgrId));
									smsSender.setResult("-1");
									smsSender.setFetch_id(flag);
									smsSender.setCity_id(cityId);
									smsSender.setRegion_id(regionId == null ? "":regionId);
									smsSender.setRegion_name(regionName == null ? "":regionName);
									smsSender.setStation_id(stationId);
									smsSender.setStation_name(stationName == null ? "":stationName); 
									smsSender.setMgr_id(mgrId);
									smsSender.setMgr_name(mgrName == null ? "":mgrName);
									smsSender.setUser_place(list.get(5) == null ? "":list.get(5));
									smsSender.setUser_city(list.get(6) == null ? "":list.get(6));
									smsSender.setIs_local(isLocal);
									smsSender.setIs_mark("0");
									smsSender.setCity_ext_id(cityExtId == null ? "":cityExtId);
									smsSender.setRedisCount(dayCount+"");
									
									String overSeven = smsTemplateAttr.getLocal_date();
									if (null != overSeven && !"".equals(overSeven)) {
										int overExpire = Integer.valueOf(overSeven)*24*60*60;
										cluster.expire(key, overExpire);
									}
									//								}
									sendToDb(smsSender,cityId,now,regionId,isLocal,key,smsTemplateAttr);
								}catch (Exception e) {
									log.error("mgrId:"+mgrId+"--变动redis中的发送量出现异常："+e.toString());
									e.printStackTrace();
								}
							} else {
								log.debug("isCz = "+isCz+",isSendLocal = "+isSendLocal+", msgId = "+mgrId);
							}
						} 
					}else {
						DataCache.todaySendMsisdns.put(msisdn+"#"+mgrId, "1");
						count3++;
						if (count3%20000 == 0) {
							log.info("redis中过滤重复数据量"+count3);
						}
					}
				}else {
					count2++;
					if (count2%20000 == 0) {
						log.info("内存中过滤重复数据量"+count2);
					}
				}
			}
		}
		/*}else{
			log.info("black user phone :"+msisdn);
		}*/
	}

	private void sendToDb(SmsSender sender,String cityId,String  now,String  regionId,String isLocal, String key,SmsTemplateAttr smsTemplateAttr) {
		senders.add(sender);
		Connection conn = null;
		int sendCount=Integer.valueOf(DealDataToSendMain.sendCount);
		try {
			if (senders.size() >= sendCount) {				
				//				log.debug("set send data to db!");
				conn = ConnectionManager.getInstance().getConnection();
				PreparedStatement pres = conn.prepareStatement(sendSql);
				for (SmsSender senderTmp: senders) {
					String msisdn = senderTmp.getReceiver();
					String tmpMgrId = senderTmp.getMgr_id();

					String tmpFlag = cluster.hget(msisdn+"#"+tmpMgrId, "INSERT_FLAG");
					cluster.incrBy(tmpMgrId, 1L);
					if (!DataCache.mgrSend.containsKey(tmpMgrId)) {
						DataCache.mgrSend.put(tmpMgrId, 1);
					} else {
						int alreadySend = DataCache.mgrSend.get(tmpMgrId);
						DataCache.mgrSend.put(tmpMgrId, alreadySend+1);
					}
					pres.setObject(1, msisdn);
					pres.setObject(2, senderTmp.getSender());
					pres.setObject(3, senderTmp.getContent());
					pres.setObject(4, senderTmp.getResult());
					pres.setObject(5, flag);
					pres.setObject(6, senderTmp.getCity_id());
					pres.setObject(7, senderTmp.getRegion_id());
					pres.setObject(8, senderTmp.getRegion_name());
					pres.setObject(9, senderTmp.getStation_id());
					pres.setObject(10, senderTmp.getStation_name());
					pres.setObject(11, tmpMgrId);
					pres.setObject(12, senderTmp.getMgr_name());
					pres.setObject(13, senderTmp.getUser_place());
					pres.setObject(14, senderTmp.getUser_city());
					pres.setObject(15, senderTmp.getIs_local());
					pres.setObject(16, senderTmp.getIs_mark());
					pres.setObject(17, senderTmp.getCity_ext_id());
					pres.setObject(18, senderTmp.getRedisCount());
					pres.addBatch();
				}
				pres.executeBatch();
				if (null != pres) {
					pres.close();
				}
				if (null != conn) {
					conn.close();
				}
				senders.clear();
				log.debug("set send data to db end!");

			}
		} catch (Exception e) {
			log.error("send data to db exception :",e);
			if (null != conn) {
				try {
					conn.close();
				} catch (SQLException e1) {
					log.error("close conn failed");
					e1.printStackTrace();
				}
			}
		}
	}


	public static void main(String[] args) {
		/*String time = "10";
		String compareTIme = "9";
		System.out.println(time.compareTo(compareTIme));*/
		
		try {
			String key = "HuaDongJieSuo123";
			String tel= "15895099536";
			System.out.println(AES.encrypt(tel, key));
		} catch (Exception e) {
			e.printStackTrace();
		}
//		ThresholdInfo info1 = new ThresholdInfo();
//		System.out.println(info1.getCycleThreshold()<10);
		
//		logger1.info("OLnwNv9hdOc1s_I6G3-EFg,Kpd5a00p0uWCN56I0I9FXg,12,20201030172017,20201030172033,2000250,12,0,0");

	}
}
